package com.pks.datacenter.job;

import java.io.File;
import java.io.FileInputStream;
import java.io.FilenameFilter;
import java.io.InputStreamReader;
import java.io.LineNumberReader;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

import org.apache.commons.lang3.StringUtils;
import org.quartz.Job;
import org.quartz.JobExecutionContext;
import org.quartz.JobExecutionException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.alibaba.fastjson.TypeReference;
import com.google.gson.JsonObject;
import com.pks.datacenter.config.RedisKeyConfig;
import com.pks.datacenter.entity.basic.DbField;
import com.pks.datacenter.entity.basic.DbInfo;
import com.pks.datacenter.entity.basic.DbSource;
import com.pks.datacenter.entity.basic.JobInfo;
import com.pks.datacenter.entity.basic.JobLog;
import com.pks.datacenter.handler.RedisHandler;
import com.pks.datacenter.listener.ApplicationContextRegister;
import com.pks.datacenter.service.basic.DbFieldService;
import com.pks.datacenter.service.basic.DbInfoService;
import com.pks.datacenter.service.basic.DbSourceService;
import com.pks.datacenter.service.basic.JobInfoService;
import com.pks.datacenter.service.basic.JobLogService;
import com.pks.datacenter.utils.SExcelReader;
import com.pks.datacenter.utils.STextFileUtil;
import com.pks.datacenter.utils.SUtilAes;
import com.pks.datacenter.utils.SUtilCommon;
import com.pks.datacenter.utils.SUtilDb;
import com.pks.datacenter.utils.SZipUtil;
import com.pks.util.UtilFile;
import com.pks.util.filenameutils.FilenameUtils;

public class QuartzJob implements Job{

//	private static SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
	private static Logger logger = LoggerFactory.getLogger(QuartzJob.class);
	
	private JobInfoService jobInfoService;
	private DbInfoService dbInfoService;
	private DbFieldService dbFieldService;
	private RedisHandler redisHandler;
	private JobLogService jobLogService;
	private DbSourceService dbSourceService;
	
	public void initService() {
		jobInfoService = (JobInfoService) ApplicationContextRegister.getApplicationContext().getBean("jobInfoService");
		dbInfoService = (DbInfoService) ApplicationContextRegister.getApplicationContext().getBean("dbInfoService");
		dbFieldService = (DbFieldService) ApplicationContextRegister.getApplicationContext().getBean("dbFieldService");
		redisHandler = (RedisHandler) ApplicationContextRegister.getApplicationContext().getBean("redisHandler");
		jobLogService = (JobLogService) ApplicationContextRegister.getApplicationContext().getBean("jobLogService");
		dbSourceService = (DbSourceService) ApplicationContextRegister.getApplicationContext().getBean("dbSourceService");
	}
	
	@Override
	public void execute(JobExecutionContext context) throws JobExecutionException {
		//任务信息上携带的信息 为jobInfo的Id，后续相关联的信息通过该表进行扩展查询
		String param = (String) context.getJobDetail().getJobDataMap().get("param");
		try {
			if(StringUtils.isNotBlank(param)){
				initService();
				Long jobId = Long.parseLong(param);
				JobInfo job = jobInfoService.findById(jobId);
				if(job!=null && JobInfo.state_on == job.getState()){
					//源表信息
					DbInfo source = dbInfoService.findById(job.getDb_source());
					//目标表信息
					DbInfo target = dbInfoService.findById(job.getDb_target());
					//目标表字段信息列表
					List<DbField> targetFields = dbFieldService.find("orders asc,id asc ", dbFieldService.initQbuilder("dbId", target.getId()), null);
					//源为数据库类型的
					if(DbInfo.db_type_mysql == source.getDb_type() || DbInfo.db_type_sqlserver == source.getDb_type() || DbInfo.db_type_oracle == source.getDb_type()){
						doDbSql(job, source, target, targetFields);
					//源为http接口的
					}else if(DbInfo.db_type_http == source.getDb_type()){
						doHttp(job, source, target, targetFields);
					//源为json文本或json文本压缩包类型的
					}else if(DbInfo.db_type_jsontxt == source.getDb_type() || DbInfo.db_type_jsontxt_zip == source.getDb_type()){
						doJsonTxt(job, source, target, targetFields);
					} else if(DbInfo.db_type_excel == source.getDb_type()) {
						doExcel(job, source, target, targetFields);
					}
				}
			}
		} catch (Exception e) {
			e.printStackTrace();
		}
	}
	
	//	处理源数据类型为 数据库的 
	public void doDbSql(JobInfo job, DbInfo sourceInfo, DbInfo targetInfo , List<DbField> targetFields) {
		try {
			String query_sql = sourceInfo.getDb_sql();
			String replaceStr = "";
			DbSource source = dbSourceService.findById(sourceInfo.getDbSourceId());
			source.setDb_password(SUtilAes.Decrypt(source.getDb_password()));
			DbSource target = dbSourceService.findById(targetInfo.getDbSourceId());
			target.setDb_password(SUtilAes.Decrypt(target.getDb_password()));
			
			if(JobInfo.type_cover == job.getType()){
				String exec_sql = "delete from `"+targetInfo.getTable_name()+"`";
				SUtilDb.executeSql(target.getDb_type(), target.getDb_ip(), target.getDb_port(), target.getDb_name(), target.getDb_account(), target.getDb_password(), exec_sql);
			}else if(JobInfo.type_add == job.getType()){
				// 根据设定的排序字段获取源数据最大值
				String exec_sql = "select max(`"+targetInfo.getOrder_field()+"`) m from `"+targetInfo.getTable_name()+"`";
				List<JSONObject> list = SUtilDb.queryList(target.getDb_type(), target.getDb_ip(), target.getDb_port(), target.getDb_name(), target.getDb_account(), target.getDb_password(), exec_sql);
				if(list.size()>0){
					JSONObject jsonObject = list.get(0);
					if(jsonObject.containsKey("m")){
						String s = String.valueOf(jsonObject.get("m"));
						// 判断排序字段类型，确定sql判断语句
						if(DbInfo.field_type_number==sourceInfo.getField_type()){
							replaceStr = " (`"+sourceInfo.getOrder_field()+"` > "+s+") ";
						}else{
							replaceStr = " (`"+sourceInfo.getOrder_field()+"` > '"+s+"') ";
						}
					}
				}
			}
			if(StringUtils.isBlank(replaceStr)){
				replaceStr = " true ";
			}
			query_sql = query_sql.replace("{sql}", replaceStr);
			List<JSONObject> list = SUtilDb.queryList(source.getDb_type(), source.getDb_ip(), source.getDb_port(), source.getDb_name(), source.getDb_account(), source.getDb_password(), query_sql);
//			List<DbField> fields = dbFieldService.find(null, dbFieldService.initQbuilder("dbId", target.getId()), null);
			SUtilDb.insertDatas(target.getDb_type(), target.getDb_ip(), target.getDb_port(), target.getDb_name(), target.getDb_account(), target.getDb_password(),targetInfo.getTable_name(), targetFields, list);
			logger.info("任务["+job.getName()+"]处理数据："+list.size()+"条");
			jobLogService.insert(new JobLog(job.getId(), job.getName(), JobLog.STATE_SUCCESS , "处理数据："+list.size()+"条"));
		} catch (Exception e) {
			e.printStackTrace();
		}
	}
	
	//	处理源数据类型为 http 请求的 
	public void doHttp(JobInfo job, DbInfo sourceInfo, DbInfo targetInfo , List<DbField> targetFields) {
		try {
			// 获取数据 参数对象
			Map<String, Object> qbuilder = new HashMap<String, Object>();
			DbSource target = dbSourceService.findById(targetInfo.getDbSourceId());
			target.setDb_password(SUtilAes.Decrypt(target.getDb_password()));
			// 如果需要鉴权进行获取鉴权值
			if (sourceInfo.getAuth_type() != null) {
				String authParamValue = null;
				if (redisHandler.hasKey(RedisKeyConfig.KEY_INFO_DBINFO_AUTHVALUE_PREFIX + sourceInfo.getId())) {
					authParamValue = String.valueOf(redisHandler.get(RedisKeyConfig.KEY_INFO_DBINFO_AUTHVALUE_PREFIX + sourceInfo.getId()));
				} else {
					if (DbInfo.AUTH_TYPE_HTTP == sourceInfo.getAuth_type()) {
						String result = SUtilCommon.httpPostRequest(sourceInfo.getAuth_url(), null);
						JSONObject jsobObject = JSONObject.parseObject(result);
						String auth_value = sourceInfo.getAuth_value();
						String authParamName = auth_value;
						if (auth_value.indexOf(".") > -1) {
							String[] auth_values = auth_value.split(".");
							for (int i = 0; i < auth_values.length; i++) {
								authParamName = auth_values[i];
								if (i < auth_values.length -1) {
									if (jsobObject.containsKey(authParamName)) {
										jsobObject = jsobObject.getJSONObject(authParamName);
									}
								}
							}
						}
						if (jsobObject.containsKey(authParamName)) {
							authParamValue = jsobObject.getString(authParamName);
						}
						if (authParamValue == null) {
							logger.error("数据源【" + sourceInfo.getName() + "】获取鉴权失败");
							jobLogService.insert(new JobLog(job.getId(), job.getName(), JobLog.STATE_FAIL , "获取鉴权失败"));
							return ;
						}  else {
							if (sourceInfo.getAuth_valid_time() != null && sourceInfo.getAuth_valid_time() > 0) {
								redisHandler.set(RedisKeyConfig.KEY_INFO_DBINFO_AUTHVALUE_PREFIX + sourceInfo.getId(), authParamValue, sourceInfo.getAuth_valid_time());
							} else {
								redisHandler.set(RedisKeyConfig.KEY_INFO_DBINFO_AUTHVALUE_PREFIX + sourceInfo.getId(), authParamValue);
							}
						}
					} else if (DbInfo.AUTH_TYPE_VALUE == sourceInfo.getAuth_type()) {
						authParamValue = sourceInfo.getAuth_value();
					}
				}
				if (authParamValue == null) {
					jobLogService.insert(new JobLog(job.getId(), job.getName(), JobLog.STATE_FAIL , "无法获得鉴权信息"));
				} else {
					qbuilder.put(sourceInfo.getAuth_param(), authParamValue);
				}
			}
			
			if(JobInfo.type_cover == job.getType()){
				String exec_sql = "delete from `"+targetInfo.getTable_name()+"`";
				SUtilDb.executeSql(target.getDb_type(), target.getDb_ip(), target.getDb_port(), target.getDb_name(), target.getDb_account(), target.getDb_password(), exec_sql);
			}else if(JobInfo.type_add == job.getType()){
				String exec_sql = "select max(`"+targetInfo.getOrder_field()+"`) m from `"+targetInfo.getTable_name()+"`";
				List<JSONObject> list = SUtilDb.queryList(target.getDb_type(), target.getDb_ip(), target.getDb_port(), target.getDb_name(), target.getDb_account(), target.getDb_password(), exec_sql);
				if(list.size()>0){
					JSONObject jsonObject = list.get(0);
					if(jsonObject.containsKey("m")){
						String s = String.valueOf(jsonObject.get("m"));
						qbuilder.put(sourceInfo.getOrder_field(), s);
					}
				}
			}
			
			
			String http_url = sourceInfo.getHttp_url();
			
			//通过http的POST请求获取源数据
			String result = SUtilCommon.httpPostRequest(http_url, SUtilCommon.paramstourl(qbuilder));
			if(StringUtils.isNotBlank(result)){
				List<JSONObject> list = new ArrayList<JSONObject>();
				// 判断是否有设置取值参数  
				if (StringUtils.isNotBlank(sourceInfo.getHttp_value_param())) {
					JSONObject resultObject = JSON.parseObject(result);
					// 判断是否需要判断获取成功
					if (StringUtils.isNotBlank(sourceInfo.getHttp_success_flag())) {
						String http_success_flag  = sourceInfo.getHttp_success_flag();
						if (http_success_flag.indexOf("=") > -1) {
							String[] flags = http_success_flag.split("=");
							if (resultObject.containsKey(flags[0]) || !resultObject.getString(flags[0]).equals(flags[1])) {
								jobLogService.insert(new JobLog(job.getId(), job.getName(), JobLog.STATE_FAIL , result));
								return ;
							}
						} else {
							if (resultObject.containsKey(http_success_flag)) {
								jobLogService.insert(new JobLog(job.getId(), job.getName(), JobLog.STATE_FAIL , result));
								return ;
							}
						}
					}
					list = JSON.parseObject(resultObject.getString(sourceInfo.getHttp_value_param()), new TypeReference<List<JSONObject>>(){});
				} else {
					list = JSON.parseObject(result,new TypeReference<List<JSONObject>>(){});
				}
				SUtilDb.insertDatas(target.getDb_type(), target.getDb_ip(), target.getDb_port(), target.getDb_name(), target.getDb_account(), target.getDb_password(),targetInfo.getTable_name(), targetFields, list);
				logger.info("任务["+job.getName()+"]处理数据："+list.size()+"条");
				jobLogService.insert(new JobLog(job.getId(), job.getName(), JobLog.STATE_SUCCESS , "处理数据："+list.size()+"条"));
			}
		} catch (Exception e) {
			e.printStackTrace();
		}
	}
	
	//	处理源数据类型为 json格式数据的txt文本或txt文本的zip的 
	public void doJsonTxt(JobInfo job, DbInfo sourceInfo, DbInfo targetInfo , List<DbField> targetFields) {
		try {
			List<String> dir_path_list = new ArrayList<String>(); 
			//判断是否需要进行删除文件夹，如果是压缩包，因为存的是解压后的文件夹需要进行删除
			boolean deleteDir = false;
			if(DbInfo.db_type_jsontxt_zip == sourceInfo.getDb_type()){
				deleteDir = true;
				//先进行压缩包文件读取
				final String rootDir = sourceInfo.getFile_path();
				File rootDirFile = new File(rootDir);
				FilenameFilter filter = new FilenameFilter() {
					@Override
					public boolean accept(File dir, String name) {
						if (name.endsWith("zip")) {
							File ftf = new File(rootDir + File.separator + FilenameUtils.getBaseName(name) + ".zip");
							if (ftf.exists()) {
								return true;
							} else {
								ftf.delete();
								UtilFile.deleteFolder(rootDir + File.separator + name);
								UtilFile.deleteFolder(rootDir + File.separator + FilenameUtils.getBaseName(name));
							}
						}
						return false;
					}
				};
				String[] zip_paths = rootDirFile.list(filter);
				if(zip_paths!=null){
					for (String zip_path : zip_paths) {
						File zip_file = new File(zip_path);
						final String dir_path = FilenameUtils.getFullPath(zip_path) + FilenameUtils.getBaseName(zip_path);
						SZipUtil.unzip(zip_file, dir_path, sourceInfo.getZip_pwd());
						zip_file.delete();
						dir_path_list.add(dir_path);
					}
				}
			}else{
				dir_path_list.add(sourceInfo.getFile_path());
			}
			//读取解压后的文件夹内的txt文件
			for (final String dir_path : dir_path_list) {
				File dir_file = new File(dir_path);
				FilenameFilter filter = new FilenameFilter() {
					@Override
					public boolean accept(File dir, String name) {
						if (name.toLowerCase().endsWith("txt")) {
							File ftf = new File(dir_path + File.separator + FilenameUtils.getBaseName(name) + ".txt");
							if (ftf.exists()) {
								return true;
							} else {
								ftf.delete();
								UtilFile.deleteFolder(dir_path + File.separator + name);
							}
						}
						return false;
					}
				};
				String[] txt_paths = dir_file.list(filter);
				List<JSONObject> list = new ArrayList<JSONObject>();
				LineNumberReader reader = null;
				try {
					for (String txt_path : txt_paths) {
						String path = dir_path +File.separator + txt_path;
						String code = STextFileUtil.resolveCode(path);
						reader = new LineNumberReader(new InputStreamReader(new FileInputStream(path), code));
						String line = null;
						while ((line = reader.readLine()) != null) {
							JSONObject jsonObject = JSONObject.parseObject(line);
							list.add(jsonObject);
						}
						File file = new File(path);
						if (file.exists()) {
							file.delete();
						}
					}
					if(deleteDir){
						SZipUtil.removeAll(dir_path);
					}
				} catch (Exception e) {
					e.printStackTrace();
				} finally {
					if (null != reader) {
						reader.close();
					}
				}

				DbSource target = dbSourceService.findById(targetInfo.getDbSourceId());
				target.setDb_password(SUtilAes.Decrypt(target.getDb_password()));
				
				SUtilDb.insertDatas(target.getDb_type(), target.getDb_ip(), target.getDb_port(), target.getDb_name(), target.getDb_account(), target.getDb_password(),targetInfo.getTable_name(), targetFields, list);
				logger.info("任务["+job.getName()+"]处理数据："+list.size()+"条");
				jobLogService.insert(new JobLog(job.getId(), job.getName(), JobLog.STATE_SUCCESS , "处理数据："+list.size()+"条"));
			}
		} catch (Exception e) {
			e.printStackTrace();
		}
		
	}
	
	//处理excel来源类型数据
	public void doExcel(JobInfo job, DbInfo sourceInfo, DbInfo targetInfo , List<DbField> targetFields){
		try {
			List<String> dir_path_list = new ArrayList<String>();
			dir_path_list.add(sourceInfo.getFile_path());
//			dir_path_list.add("F:/excel");
			//读取解压后的文件夹内的txt文件
			for (final String dir_path : dir_path_list) {
				File dir_file = new File(dir_path);
				FilenameFilter filter = new FilenameFilter() {
					@Override
					public boolean accept(File dir, String name) {
						if (name.toLowerCase().endsWith("xls")) {
							File ftf = new File(dir_path + File.separator + FilenameUtils.getBaseName(name) + ".xls");
							if (ftf.exists()) {
								return true;
							} else {
								ftf.delete();
								UtilFile.deleteFolder(dir_path + File.separator + name);
							}
						}
						return false;
					}
				};
				String[] files = dir_file.list(filter);
				List<JSONObject> list = new ArrayList<JSONObject>();
				for (String filepath : files) {
					File file = new File(dir_file + File.separator + filepath);
					List<List<String>> dataList = SExcelReader.readXls(new FileInputStream(file), 0, null);
					list.addAll(convertData(dataList, targetFields));
					if (file.exists()) {
						file.delete();
					}
				}
				
				DbSource target = dbSourceService.findById(targetInfo.getDbSourceId());
				target.setDb_password(SUtilAes.Decrypt(target.getDb_password()));
				
				SUtilDb.insertDatas(target.getDb_type(), target.getDb_ip(), target.getDb_port(), target.getDb_name(), target.getDb_account(), target.getDb_password(),targetInfo.getTable_name(), targetFields, list);
				logger.info("任务["+job.getName()+"]处理数据："+list.size()+"条");
				jobLogService.insert(new JobLog(job.getId(), job.getName(), JobLog.STATE_SUCCESS , "处理数据："+list.size()+"条"));
			}
		} catch (Exception e) {
			e.printStackTrace();
		}
	}

	
	//处理数据格式
	private static List<JSONObject> convertData(List<List<String>> dataList,List<DbField> targetFields){
		List<JSONObject> resultList = new ArrayList<JSONObject>();
		for (List<String> data : dataList) {
			JSONObject jsonObject = new JSONObject();
			for (int i = 0; i < targetFields.size(); i++) {
				if (data.size() > i) {
					break ;
				}
				jsonObject.put(targetFields.get(i).getEnName(), data.get(i));
			}
			resultList.add(jsonObject);
		}
		return resultList;
	}
	
}
